以下參考課程 LLM Twin: Building Your Production-Ready AI Replica 撰寫

在探討完設計模式後，我們將聚焦第三堂課中提及的 Change Data Capture (CDC) 技術。在構建 GenAI 服務時，由於來源資料會不斷更新，比如在課程案例中，我們需要確保來自 LinkedIn、Medium 和 GitHub 等多個平台的資料和資料庫保持同步和最新，而 CDC 正是解決這一問題的方法。

接下來的三天，我將逐步由淺入深地介紹 CDC 技術。

CDC簡介

Change Data Capture (CDC) 是一種用於捕捉資料庫變更的技術。CDC 的主要功能包括：

• 識別資料變更：監控並記錄資料庫（如 MongoDB）的插入、更新和刪除操作。
• 生成變更日誌：將捕捉到的變更轉換為結構化的日誌。
• 傳播變更訊息：將變更事件發送到消息隊列（如 RabbitMQ），供其他系統處理。

CDC 在現代數據系統中的應用非常廣泛。首先，它通過實時同步變更至目標資料庫，確保服務之間的一致性。其次，CDC 只處理變更數據，優化了系統效能。此外，CDC 支援事件驅動架構，將數據變更視為觸發系統操作的事件，統一接收端介面。

在分佈式系統中，CDC 解決了一致性問題：它解耦了資料庫更新和消息發送，將資料庫的變動作為觸發事件，即使應用程序故障，也能可靠地發送消息。這些特性提升了系統的靈活性、效率和可靠性，使 CDC 特別適合複雜的分佈式環境。

CDC 的工作流程

透過課程的圖片，我們可以拆解 CDC 的工作流程

1. 沒有 CDC 的情況 （左上橘黃色部份）
   1. 應用程式觸發：透過 REST 調用或事件來觸發操作。
   2. 更新資料庫：應用程式更新資料庫。
   3. 資料庫確認：資料庫向應用程式確認更新完成。
   4. 發送訊息：應用程式嘗試透過消息服務發送訊息（如 Kafka）。
   5. 在沒有 CDC 的情況下，如果應用程式在步驟3完成、步驟4尚未執行完畢時失敗，可能會導致數據不一致。
2. 有 CDC 的情況 （右下藍色部份）
   1. CDC 監控數據庫的變更日誌（如 MongoDB 的 oplog）。
   2. 當發現變更時，CDC 自動捕捉這個事件。
   3. CDC 將變更事件發送到消息服務（如 Kafka 或 RabbitMQ）。
   4. 其他系統訂閱這些消息，保持數據一致性。

CDC 在 GenAI 系統中的應用

在 GenAI 系統中，特別是在實現**檢索增強生成（RAG）**時，CDC 扮演著關鍵角色：

1. 知識庫實時更新
   • 確保 GenAI 系統能夠使用最新的知識來回答問題。
   • CDC 解決方案：
     • 監控資料庫的變更。
     • 當檢測到變更時，自動生成新文檔的文本向量。
     • 將新生成的向量更新到向量數據庫中。

# 偽代碼範例
def on_document_change(change_event):
    doc = change_event['fullDocument']
    embeddings = generate_embeddings(doc['content'])
    vector_db.upsert(doc['id'], embeddings)

cdc_stream.watch(on_document_change)

2. 訓練數據同步
   • 確保模型訓練過程能及時利用新的訓練數據。
   • CDC 解決方案：
     • 監控訓練資料的變更。
     • 當增加新的訓練資料時，將其推送到訓練隊列。
     • 當訓練資料累積到一定閾值時，自動觸發模型微調過程。

def on_training_data_change(change_event):
    data = change_event['fullDocument']
    training_queue.push(data)
    if training_queue.size > THRESHOLD:
        trigger_model_fine_tuning()

cdc_stream.watch(on_training_data_change)

3. 多模態數據整合
   • 目的：保持多模態數據（如圖片、音頻）與其文本描述的同步。
   • CDC 解決方案：
     • 監控多模態資料（如圖片）的變更。
     • 當檢測到新的或更新的多模態數據時，自動生成或更新其文字描述。
     • 將生成的描述更新到資料庫中。

def on_image_change(change_event):
    image = change_event['fullDocument']
    description = generate_image_description(image['url'])
    text_db.update(image['id'], {'description': description})

cdc_stream.watch(on_image_change)